package com.f.clickhouse;

import cn.hutool.core.bean.BeanUtil;
import cn.hutool.core.date.LocalDateTimeUtil;
import cn.hutool.core.util.StrUtil;
import com.clickhouse.client.*;
import com.clickhouse.data.ClickHouseFormat;
import com.clickhouse.data.ClickHouseRecord;
import com.f.constant.Constant;
import com.google.common.base.CaseFormat;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.util.CollectionUtils;
import org.springframework.util.StopWatch;

import java.io.Serializable;
import java.lang.reflect.InvocationTargetException;
import java.time.LocalDateTime;
import java.util.*;
import java.util.concurrent.CompletableFuture;
import java.util.stream.Collectors;

/**
 * clickhouse Template
 *
 * @author liuf
 * @date 2023/6/15 15:30
 */
public class ClickHouseTemplate {

    private static final Logger LOGGER = LoggerFactory.getLogger(ClickHouseTemplate.class);

    private final ClickHouseClient clickHouseClient;

    private final ClickHouseNodes clickHouseNodes;

    private final FieldToColumn fieldToColumn;

    public ClickHouseTemplate(ClickHouseClient clickHouseClient, ClickHouseNodes clickHouseNodes, FieldToColumn fieldToColumn) {
        this.clickHouseClient = clickHouseClient;
        this.clickHouseNodes = clickHouseNodes;
        if (fieldToColumn == null) {
            fieldToColumn = column -> CaseFormat.LOWER_CAMEL.to(CaseFormat.LOWER_UNDERSCORE, column);
        }
        this.fieldToColumn = fieldToColumn;
    }

    public ClickHouseRequest<?> getRequest() {
        return clickHouseClient.read(clickHouseNodes);
    }

    public ClickHouseResponse query(String sql, Object[] params) {
        final StopWatch stopWatch = new StopWatch();
        try {
            stopWatch.start();
            return getRequest().format(ClickHouseFormat.RowBinaryWithNamesAndTypes)
                    .query(sql).params(params).executeAndWait();
        } catch (ClickHouseException e) {
            throw new RuntimeException(e);
        } finally {
            stopWatch.stop();
            LOGGER.info("query time:{}", stopWatch.getLastTaskTimeMillis());
        }
    }

    public CompletableFuture<ClickHouseResponse> queryAsync(String sql, Object[] params) {
        return getRequest().format(ClickHouseFormat.RowBinaryWithNamesAndTypes)
                .query(sql).params(params).execute();
    }

    public Iterable<ClickHouseRecord> select(String sql, Object[] params) {
        try (final ClickHouseResponse response = query(sql, params)) {
            return response.records();
        }
    }

    public <T> List<T> selectList(String sql, Class<T> clazz, Object[] params) {
        try (final ClickHouseResponse response = query(sql, params)) {
            return response.stream(clazz).collect(Collectors.toList());
        }
    }

    public <T> T selectFirst(String sql, Class<T> clazz, Object[] params) {
        try (final ClickHouseResponse response = query(sql, params)) {
            return response.firstRecord(clazz);
        }
    }

    public int insertSql(String sql, Object[] params) {
        try (final ClickHouseResponse response = query(sql, params)) {
            return ((int) response.getSummary().getWrittenRows());
        }
    }

    public int insert(String table, String[] fields, Map<String, Object> params) {
        if (params == null || params.isEmpty() || fields == null || fields.length == 0) {
            return 0;
        }

        Object[] objects = new Object[fields.length];
        for (int i = 0; i < fields.length; i++) {
            objects[i] = params.get(fields[i]);
        }
        return insertSql(prepInsertSql(table, getColumns(fields)), objects);
    }

    public int insert(String table, String[] fields, Object firstParam, Object... params) {
        Object[] objects = new Object[fields.length];
        objects[0] = firstParam;
        System.arraycopy(params, 0, objects, 1, fields.length - 1);
        return insertSql(prepInsertSql(table, getColumns(fields)), objects);
    }

    private List<String> getColumns(String[] fields) {
        return Arrays.stream(fields).map(fieldToColumn::toColumn).collect(Collectors.toList());
    }

    public int batchInsert(String table, String[] fields, List<Map<String, Object>> list) {
        if (CollectionUtils.isEmpty(list)) {
            return 0;
        }
        final int size = list.size();
        Object[] params = new Object[size * fields.length];
        for (int i = 0; i < size; i++) {
            final Map<String, Object> map = list.get(i);
            for (int j = 0, len = fields.length; j < len; j++) {
                params[i * len + j] = map.get(fields[j]);
            }
        }
        return insertSql(prepBatchInsertSql(table, getColumns(fields), size), params);
    }

    public <T extends Serializable> int batchInsert(String table, List<T> list, String[] fields) {
        if (CollectionUtils.isEmpty(list)) {
            return 0;
        }
        final StopWatch stopWatch = new StopWatch();
        stopWatch.start();
        final int size = list.size();
        Object[] params = new Object[size * fields.length];
        Object value;
        for (int i = 0; i < size; i++) {
            for (int j = 0, len = fields.length; j < len; j++) {
                value = BeanUtil.getFieldValue(list.get(i), fields[j]);
                if (value instanceof LocalDateTime) {
                    value = LocalDateTimeUtil.format((LocalDateTime) value, Constant.DATE_TIME_FORMAT);
                }
                params[i * len + j] = value;
            }
        }
        stopWatch.stop();
        LOGGER.info("set field time:{}", stopWatch.getLastTaskTimeMillis());
        return insertSql(prepBatchInsertSql(table, getColumns(fields), size), params);
    }

    public int insert(String table, Object object) {
        if (object == null) {
            return 0;
        }
        final Map<String, Object> map = BeanUtil.beanToMap(object, false, true);
        return insert(table, map, map.keySet().toArray(new String[0]));
    }

    public int insert(String table, Object object, String[] fields) {
        if (object == null) {
            return 0;
        }
        Object[] params = new Object[fields.length];
        Object value;
        for (int i = 0, len = fields.length; i < len; i++) {
            value = BeanUtil.getFieldValue(object, fields[i]);
            if (value instanceof LocalDateTime) {
                value = LocalDateTimeUtil.format(((LocalDateTime) value), Constant.DATE_TIME_FORMAT);
            }
            params[i] = value;
        }
        return insertSql(prepInsertSql(table, getColumns(fields)), params);
    }

    public <T extends Serializable> int batchInsert(String table, List<T> list) {
        if (CollectionUtils.isEmpty(list)) {
            return 0;
        }
        Map<String, Object> map = BeanUtil.beanToMap(list.get(0));
        return batchInsert(table, list, map.keySet().toArray(new String[]{}));
    }

    private static String prepInsertSql(String table, Collection<String> columns) {
        final String sql = new StringJoiner(" ")
                .add("insert into")
                .add(table)
                .add("(")
                .add(String.join(",", columns))
                .add(")")
                .add("values")
                .add("(")
                .add(columns.stream().map(key -> ":" + key).collect(Collectors.joining(",")))
                .add(")")
                .toString();
        LOGGER.info("sql:{}", sql);
        return sql;
    }

    private static String prepBatchInsertSql(String table, Collection<String> columns, int size) {
        final StringJoiner stringJoiner = new StringJoiner(" ")
                .add("insert into")
                .add(table)
                .add("(")
                .add(String.join(",", columns))
                .add(")")
                .add("values");
        for (int i = 0; i < size; i++) {
            String ci = String.valueOf(i);
            stringJoiner.add("(")
                    .add(columns.stream().map(key -> ":" + key + ci).collect(Collectors.joining(",")));
            if (i < size - 1) {
                stringJoiner.add("),");
            } else {
                stringJoiner.add(")");
            }
        }
        String sql = stringJoiner.toString();
        LOGGER.info("sql:{}", StrUtil.subPre(sql, 256));
        return sql;
    }

    public void destroy() {
        LOGGER.info("destroy");
        clickHouseClient.close();
        clickHouseNodes.shutdown();
    }

    public static void main(String[] args) throws InvocationTargetException, IllegalAccessException {

    }
}
